package org.example.jobs;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.thread.NamedThreadFactory;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.simple.job.SimpleJob;
import org.example.config.TaskParam;
import org.example.entity.Test;
import org.example.entity.TestData;
import org.example.entity.TestJobData;
import org.example.jobs.job.BatchDataRunnable;
import org.example.service.*;
import org.example.util.JobUtil;
import org.example.util.ListQueue;
import org.quartz.DisallowConcurrentExecution;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author yangzhenyu
 * @version 1.0
 * @description:
 * @date 2023/4/26 14:33
 */
@Slf4j
@Component
@DisallowConcurrentExecution
public class TestJobMany implements SimpleJob {
    private static final int LIMIT_PAGE = 500;

    @Override
    public void execute(ShardingContext context) {
        //作业名称
        String jobName = context.getJobName();
        //分片序列号
        int shardingItem = context.getShardingItem();
        //分片序列号对应的value值
        String shardingParameter = context.getShardingParameter();
        //作业分片总数
        int shardingTotalCount = context.getShardingTotalCount();
        long batchStart = System.currentTimeMillis();
        List<String> errorData = Lists.newArrayList();

        log.info("{} begin, 作业分片总数: {}, 分片序列号: {},分片序列号对应的value值：{}",  jobName,shardingTotalCount, shardingItem,shardingParameter);
        TaskParam taskParam = SpringUtil.getBean(TaskParam.class);
        int curHour = LocalDateTime.now().getHour();
        //来源数据
        TestService testService = SpringUtil.getBean(TestService.class);
        JobService jobService = SpringUtil.getBean(JobService.class);
        TestJobService testJobService = SpringUtil.getBean(TestJobService.class);
        //线程池
        ThreadPoolExecutor executorService = SpringUtil.getBean("jobThreadPoolExecutor");
        //查看时候执行完毕
        if(0 < testJobService.queryCount(shardingItem)){
            return;
        }
        //查询最大id
        Long maxId = testService.queryMaxId();
        //开始id
        long beginId = 0;
        // 如果不在规定的时间范围内，那么任务结束
        if (!JobUtil.ifCurrentHourInRange(curHour, taskParam.getStartHour(), taskParam.getEndHour())) {
            log.info("task end, 作业分片总数: {}, 分片序列号: {}, 当前时间：{}, 允许任务执行的开始时间：{},允许任务执行的结束时间:{}",
                    shardingTotalCount, shardingItem, curHour,taskParam.getStartHour(), taskParam.getEndHour());
            return;
        }
        // 只要开始id  <= 最大id+批次，则开始执行
        while (beginId <= maxId + LIMIT_PAGE){
            List<Test> query = testService.query(beginId, shardingItem, shardingTotalCount, LIMIT_PAGE);
            if (CollectionUtil.isNotEmpty(query)) {
                Future<List<Integer>> submit = executorService.submit(new BatchDataRunnable(jobService, query));
                ListQueue.getQueue().addLast(submit);
                Test test = query.get(query.size() - 1);
                beginId = test.getId() +1;
            }else {
                beginId += LIMIT_PAGE;
            }
        }
        try {
            //处理线程池回调数据
            while (0 != ListQueue.getQueue().size()){
                Future<List<Integer>> first = ListQueue.getQueue().getFirst();
                if (first.isDone()) { //保证任务执行完删除队列
                    List<Integer> integers = first.get();
                    if (0 == integers.size()) {
                        errorData.add(JobUtil.classToJson(integers));
                    }
                    //这里抓取数据或做业务处理
                    ListQueue.getQueue().removeFirst();//在队列里删除该结果
                }
            }
            }catch (Exception e){
                log.error("整理错误数据异常",e);
            }
        if (CollectionUtil.isNotEmpty(errorData)){
            List<TestJobData> list = Lists.newArrayList();
            for (String data:errorData){
                TestJobData build = TestJobData.builder()
                        .errorData(data)
                        .shardingItem(shardingItem)
                        .build();
                list.add(build);
            }
            jobService.saveBatch(list);
        }
        testJobService.add(shardingItem);
       log.info("use time: {}, 作业分片总数: {}, 分片序列号: {}, beginId: {}",
                (System.currentTimeMillis() - batchStart), shardingTotalCount, shardingItem, beginId);


    }
}
